package cn.bearspark.framework.mq.starter.config;

import cn.bearspark.framework.mq.starter.properties.RocketMQProducerProperties;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * RocketMQ 的生产者的配置类
 */
@Configuration
public class RocketMQProducerConfig {

    /**
     * 创建并配置一个生产者实例
     *
     * @param rocketMQProducerProperties RocketMQ 生产者的配置属性
     * @return 配置好的生产者实例
     * @throws Exception 启动生产者时发生的异常
     */
    @Bean
    public MQProducer mqProducer(RocketMQProducerProperties rocketMQProducerProperties) throws Exception {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        // 设置 nameSrv 的地址
        defaultMQProducer.setNamesrvAddr(rocketMQProducerProperties.getNameSrv());
        // 设置 生产者组的组名
        defaultMQProducer.setProducerGroup(rocketMQProducerProperties.getGroupName());
        // 设置 消息发送失败时可以重试另一个 broker
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        // 设置 发送消息的超时时间
        defaultMQProducer.setSendMsgTimeout(rocketMQProducerProperties.getSendMsgTimeout());
        // 设置 消息发送失败的重试次数
        defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProducerProperties.getRetryTimes());
        // 设置 异步消息发送失败的重试次数
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(rocketMQProducerProperties.getRetryTimes());
        // 设置 异步消息发送的线程池
        defaultMQProducer.setAsyncSenderExecutor(new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors() * 2,                     // 核心线程数为 可用处理器数 * 2
                100,                                                                // 最大线程数为 100
                30, TimeUnit.SECONDS,                                               // 空闲线程存活时间为 30s
                new ArrayBlockingQueue<>(2000),                                     // 任务队列为大小 2000 的有界阻塞队列
                r -> new Thread(r, "RocketMQ-async-thread-" + new Random().ints())  // 给每个线程设置名称
        ));
        defaultMQProducer.start();
        return defaultMQProducer;
    }
}
